package com.hzg.stream.watermark;

import com.hzg.bean.Record;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author: huangzhigao
 * @Date: 2022/4/18 19:52
 */
public class WaterMarkDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<Record> records = new ArrayList<>();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStreamSource<Record> source = env.fromCollection(records);
        SerializableTimestampAssigner<Record> recordTimestampAssigner = (SerializableTimestampAssigner<Record>) (element, recordTimestamp) -> element.getRecordTime();
//        source.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenertor(Time.of(3, TimeUnit.SECONDS)));
        source.assignTimestampsAndWatermarks(WatermarkStrategy.<Record>forBoundedOutOfOrderness(Duration.ofMinutes(3)).withTimestampAssigner(recordTimestampAssigner));

    }


    public static class BoundedOutOfOrdernessGenertor extends BoundedOutOfOrdernessTimestampExtractor<Record> {

        public BoundedOutOfOrdernessGenertor(Time maxOutOfOrderness) {
            super(maxOutOfOrderness);
        }

        @Override
        public long extractTimestamp(Record element) {
            return element.getRecordTime();
        }
    }
}
